package com.lagou.rpc.common.handler;

import com.lagou.rpc.common.entity.SimpleRpcProtocol;
import com.lagou.rpc.common.enums.ProtocolCheckResultEnum;
import com.lagou.rpc.common.exceptions.SimpleRpcException;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;
import java.util.function.Function;

/**
 * simpleRpc协议解析器
 *
 * @author wlz
 * @date 2020/4/30
 */
public class SimpleRpcProtocolDecoder extends ByteToMessageDecoder {


    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        byteBuf.markReaderIndex();
        Object result = new ReadChain(byteBuf)
                // 读取协议标记
                .read(this::readProtocolMark)
                // 读取请求头部
                .read(this::readHeader)
                // 读取内容
                .read(this::readContent)
                // 处理结果
                .over(channelHandlerContext);
        // 将解析到的结果放入list中
        if (result != null) {
            list.add(result);
        }
    }

    private HandleResult readProtocolMark(Target target) {
        int protocolMarkLength = SimpleRpcProtocol.SIMPLE_RPC_MARK.length();
        ByteBuf byteBuf = target.byteBuf;
        if (byteBuf.readableBytes() < protocolMarkLength) {
            return unfinish();
        }
        byte[] bytes = new byte[protocolMarkLength];
        // 读取协议标志
        byteBuf.readBytes(bytes);
        String mark = new String(bytes);
        // 如果不同则返回错误
        if (!SimpleRpcProtocol.SIMPLE_RPC_MARK.equals(mark)) {
            return error(new SimpleRpcException("invalid rpc mark"));
        }
        // 如果相同，则设置标记，并通知下一个读取
        target.simpleRpcProtocol.setSimpleRpcMark(mark);
        return next();
    }

    private HandleResult readHeader(Target target) {
        ByteBuf byteBuf = target.byteBuf;
        if (byteBuf.readableBytes() < 2) {
            return unfinish();
        }
        // 读取header长度
        short headerLength = byteBuf.readShort();
        if (byteBuf.readableBytes() < headerLength) {
            byteBuf.resetReaderIndex();
            return unfinish();
        }
        // 读取header内容
        byte[] headerContent = new byte[headerLength];
        byteBuf.readBytes(headerContent);
        target.simpleRpcProtocol.setHeaderLength(headerLength);
        target.simpleRpcProtocol.setHeader(headerContent);
        return next();
    }

    private HandleResult readContent(Target target) {
        ByteBuf byteBuf = target.byteBuf;
        if (byteBuf.readableBytes() < 4) {
            return unfinish();
        }
        // 读取content长度
        int contentLength = byteBuf.readInt();
        if (byteBuf.readableBytes() < contentLength) {
            byteBuf.resetReaderIndex();
            return unfinish();
        }
        // 读取content内容
        byte[] content = new byte[contentLength];
        byteBuf.readBytes(content);
        target.simpleRpcProtocol.setLength(contentLength);
        target.simpleRpcProtocol.setContent(content);
        return success();
    }

    private static class ReadChain {
        private boolean finish;
        private Target target;
        private HandleResult result;


        public ReadChain(ByteBuf byteBuf) {
            finish = false;
            this.target = new Target(new SimpleRpcProtocol(), byteBuf);
        }

        public ReadChain read(Function<Target, HandleResult> protocolHandler) {
            if (finish) {
                return this;
            }
            result = protocolHandler.apply(target);
            switch (result.resultType) {
                case UNFINISH:
                case FAILED:
                case SUCCESS:
                    finish = true;
                    break;
                default:
                    break;
            }
            return this;
        }

        public Object over(ChannelHandlerContext channelHandlerContext) {
            switch (result.resultType) {
                case UNFINISH:
                    return null;
                case SUCCESS:
                    // 协议读取成功,直接返回协议对象
                    return target.simpleRpcProtocol;
                case FAILED:
                    // 如果协议解析失败，发送错误消息，然后直接断开连接
                    channelHandlerContext.writeAndFlush(result.error.getMessage());
                    channelHandlerContext.disconnect().addListener(ChannelFutureListener.CLOSE);
                    break;
                case NEXT:
                default:
                    throw new SimpleRpcException("operation illegal!");
            }
            return null;
        }


    }

    private static class Target {
        private SimpleRpcProtocol simpleRpcProtocol;
        private ByteBuf byteBuf;

        public Target(SimpleRpcProtocol simpleRpcProtocol, ByteBuf byteBuf) {
            this.simpleRpcProtocol = simpleRpcProtocol;
            this.byteBuf = byteBuf;
        }
    }

    private static class HandleResult {
        private ProtocolCheckResultEnum resultType;
        private Throwable error;
    }

    private HandleResult success() {
        HandleResult handleResult = new HandleResult();
        handleResult.resultType = ProtocolCheckResultEnum.SUCCESS;
        return handleResult;
    }

    private HandleResult error(Throwable th) {
        HandleResult handleResult = new HandleResult();
        handleResult.resultType = ProtocolCheckResultEnum.FAILED;
        handleResult.error = th;
        return handleResult;
    }

    private HandleResult next() {
        HandleResult handleResult = new HandleResult();
        handleResult.resultType = ProtocolCheckResultEnum.NEXT;
        return handleResult;
    }

    private HandleResult unfinish() {
        HandleResult handleResult = new HandleResult();
        handleResult.resultType = ProtocolCheckResultEnum.UNFINISH;
        return handleResult;
    }
}
